package com.alan.mr.weibo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * Created by Alan on 2017/10/2.
 */
public class SecondJob {

    public static void main(String[] args) {
        Configuration config = new Configuration();
        config.set("fs.default", "hdfs://192.168.208.106:8020");
        config.set("yarn.resourcemanager.hostname", "192.168.208.106");

        try {

            Job job = Job.getInstance(config);
            job.setJarByClass(SecondJob.class);
            job.setJobName("second");

            job.setMapperClass(SecondMapper.class);
            job.setCombinerClass(SecondReducer.class);
            job.setReducerClass(SecondReducer.class);

            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);

            //FirstJob的输出为输入
            FileInputFormat.addInputPath(job, new Path("E:\\sxt\\1-MapReduce\\data\\weibo1"));

            FileSystem fileSystem = FileSystem.get(config);
            Path outPath = new Path("E:\\sxt\\1-MapReduce\\data\\weibo2");
            if (fileSystem.exists(outPath)) {
                fileSystem.delete(outPath);
            }
            FileOutputFormat.setOutputPath(job, outPath);

            boolean f = job.waitForCompletion(true);
            if (f) {
                System.out.println("job2 success");
            }

        }catch(Exception e) {
            e.printStackTrace();
        }

    }
}
